package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constans.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * kafkastreams流式处理
 */
@Component
public class HotArticleStreamHandle {
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //1.接收消息内容
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2.流式计算
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 初始化接收的消息（重新定义key和value）
             * @param key 表示发送消息指定的key,此时key=null
             * @param value 表示发送消息的内容，字符串类型，包含的就是UpdateArticleMess对象的字符串json化
             * @return key=文章id,value= type:add
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1把value转成对象
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                return new KeyValue<>(mess.getArticleId().toString(),mess.getType()+":"+mess.getAdd());
            }
        })
                /**
                 *  分组，根据文章id进行分组，每个文章就是一个组
                 *  key=文章id
                 *  value=  type:add 比如 value= likes:1
                 */
                .groupBy((key,value)-> key)
                //可以处理晚到消息，每10s中处理一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始化
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                },
                        new Aggregator<String, String, String>() {
                    /**
                     * 真正的聚合处理
                     * @param key =文章id
                     * @param value =type:add
                     * @param init="COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                     * @return  "COLLECTION:1,COMMENT:2,LIKES:3,VIEWS:1"
                     */
                    @Override
                    public String apply(String key, String value, String init) {
                        //1.判断value是否等于空
                        if(value==null){
                            return init;
                        }
                        //定义初始化变量
                        Integer collection = 0,comment=0,like=0,view=0;

                        //2.分割value,并且对每种类型的数值进行增量
                        String[] split = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                            case LIKES:
                                like+=Integer.parseInt(split[1]);
                                break;
                            case VIEWS:
                                view+=Integer.parseInt(split[1]);
                                break;
                            case COMMENT:
                                comment+=Integer.parseInt(split[1]);
                                break;
                            case COLLECTION:
                                collection+=Integer.parseInt(split[1]);
                                break;
                        }
                        String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);
                        return format;
                    }
                },
                        /**
                         * 聚合之后的名称，随意定义
                         */
                        Materialized.as("myAgg"))
                .toStream()//转化类型为kstream类型
                //把处理之后的消息内容进行封装
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String,String>>() {
                    /**
                     * 处理的消息之后的封装
                     * @param key =文章id
                     * @param value="COLLECTION:1,COMMENT:2,LIKES:3,VIEWS:4"  消息内容，但是缺少文章id
                     * @return
                     */
                    @Override
                    public KeyValue<String,String> apply(Windowed<String> key, String value) {
                        return new KeyValue<String,String>(key.key(),formatMess(key.key(),value));
                    }
                })
                //3.发送消息内容
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);//发送到指定的topic中
        return kStream;
    }

    /**
     * 封装消息内容
     * @param key 文章id
     * @param value "COLLECTION:1,COMMENT:2,LIKES:3,VIEWS:4"
     * @return 包含文章id,点赞，收藏，阅读，评论这个五个字段
     */
    private String formatMess(String key, String value) {
        ArticleVisitStreamMess mess=new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(key));
        String[] split = value.split(",");
        for (String agg : split) {
            String[] aggsplit = agg.split(":");//aggsplit[0]=collection,aggsplit[1]=1
            switch (UpdateArticleMess.UpdateArticleType.valueOf(aggsplit[0])){
                case LIKES:
                   mess.setLike(Integer.parseInt(aggsplit[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(aggsplit[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(aggsplit[1]));
                    break;
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(aggsplit[1]));
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }
}
